package com.elitel.frame.business.service.impl;

import com.elitel.common.metadata.*;
import com.elitel.common.mongodb.MongoDBUtil;
import com.elitel.common.mongodb.QueryInfo;
import com.elitel.frame.base.service.HiveService;
import com.elitel.frame.base.service.KuduService;
import com.elitel.frame.business.dao.ext.DataBaseMetadataMapper;
import com.elitel.frame.business.dao.ext.ExecuteSqlMapper;
import com.elitel.frame.business.service.DataBaseMetadataService;
import com.elitel.frame.core.service.impl.BaseServiceImpl;
import com.elitel.frame.main.dao.ext.ExtCfgTableMapper;
import com.elitel.frame.main.entity.CfgDb;
import com.elitel.frame.main.entity.CfgTable;
import com.elitel.frame.main.entity.DataBaseType;
import com.elitel.frame.main.entity.DataSourceType;
import com.elitel.frame.main.entity.constant.DictionaryConstant;
import com.elitel.frame.main.entity.vo.KeyValueModelVo;
import com.mongodb.client.MongoDatabase;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author gyf @Description 数据库表、字段元数据服务接口实现类
 * @data 2017-07-31
 */
@Service
public class DataBaseMetadataServiceImpl extends BaseServiceImpl implements DataBaseMetadataService {
    private static final Logger logger = LoggerFactory.getLogger(DataBaseMetadataServiceImpl.class);

    @Autowired(required = false)
    private ExecuteSqlMapper executeSqlMapper;
    @Autowired(required = false)
    private DataBaseMetadataMapper dataBaseMetadataMapper;
    @Autowired(required = false)
    private KuduService kuduService;
    @Autowired(required = false)
    private HiveService hiveService;

    @Autowired
    private ExtCfgTableMapper extCfgTableMapper;
    @Autowired
    Environment environment;

    @Autowired
    @Qualifier("slaveDataSource")
    DataSource dataSource;

    /**
     * MySQL 数据库标识
     */
    private final static String MYSQL_DIALECT = "MYSQL";

    /**
     * 获取所有表名通过数据源驱动包
     *
     * @param cfgdb
     * @return
     */
    public List<Map<String, String>> getAllTableByDriver(CfgDb cfgdb) {
        List<Map<String, String>> resultList = new ArrayList<>();
        try {
            //获取所有表
            List<MetaTable> tables = DatabaseMetaDataHelper.getTables(dataSource.getConnection(), getCurrentSchema(cfgdb),
                    new String[]{DatabaseMetaDataHelper.TableTypeEnum.TABLE.initValue()});
            tables.forEach(map -> {
                Map<String, String> dataMap = new LinkedHashMap<>();
                dataMap.put("table_name", map.getTableName());
                dataMap.put("comments", map.getRemarks());
                resultList.add(dataMap);
            });
        } catch (SQLException sqlException) {
            logger.info("获取数据库信息异常，请检查调用参数。", sqlException);
        }
        return resultList;
    }

    @Override
    public List<Map<String, String>> getAllTableByDataSource(CfgDb cfgdb) {
        List<Map<String, String>> listResult = new ArrayList<Map<String, String>>();
        if (DataBaseType.MongoDB.equals(cfgdb.getDbVersion())) {
            MongoDBUtil instance = MongoDBUtil.getInstance();
            MongoDatabase connection = instance.createConnection(cfgdb.getDbUrl());
            List<String> collections = instance.getCollections(connection);
            if (collections.size() > 0) {
                for (String name : collections) {
                    Map<String, String> map = new HashMap<>();
                    map.put("table_name", name);
                    map.put("comments", "");
                    listResult.add(map);
                }
            }
            instance.close();
        } else if (DataBaseType.KUDU.equals(cfgdb.getDbVersion())) {
            List<String> allTables = kuduService.getAllTables(cfgdb.getDbName());
            listResult = allTables.stream().map(o -> {
                return new HashMap<String, String>() {{
                    put("table_name", o);
                    put("comments", "");
                }};
            }).collect(Collectors.toList());
        } else {
            return getAllTableByDriver(cfgdb);
        }
        return listResult;
    }

    /**
     * 获取主键值通过数据源驱动包
     *
     * @param cfgdb
     * @param tableName
     * @return
     */
    public String getPrimaryKeysByDriver(CfgDb cfgdb, String tableName) {
        try {
            //获取主键信息
            List<MetaPrimaryKey> primaryKeys = DatabaseMetaDataHelper.getPrimaryKeys(dataSource.getConnection(),
                    getCurrentSchema(cfgdb), tableName);
            if (!ObjectUtils.isEmpty(primaryKeys)) {
                return primaryKeys.stream().map(MetaPrimaryKey::getColumnName).collect(Collectors.joining(","));
            }
        } catch (SQLException sqlException) {
            logger.info("获取数据库信息异常，请检查调用参数。", sqlException);
        }
        return null;
    }

    @Override
    public String getPkName(CfgDb cfgdb, String tableName) {
        if (DataBaseType.MongoDB.equals(cfgdb.getDbVersion())) {
//            List<String> pklist = dataBaseMetadataMapper.getPkNameDM8(tableName.substring(tableName.indexOf(".") + 1),
//                    tableName.substring(0, tableName.indexOf(".")));
//            return String.join(",", pklist);
            return "_id";
        } else if (DataBaseType.KUDU.equals(cfgdb.getDbVersion())) {
            List<String> primaryKey = kuduService.getPrimaryKey(cfgdb.getDbName(), tableName);
            return String.join(",", primaryKey);
        } else {
            return getPrimaryKeysByDriver(cfgdb, tableName);
        }
    }

    /**
     * 获取所有表名通过数据源驱动包
     *
     * @param cfgdb
     * @param tableName
     * @return
     */
    public List<KeyValueModelVo> getColumnsByDriver(CfgDb cfgdb, String tableName) {
        List<KeyValueModelVo> resultList = new ArrayList<>();
        try {
            //获取所有列名
            List<MetaColumn> tables = DatabaseMetaDataHelper.getColumns(dataSource.getConnection(),
                    getCurrentSchema(cfgdb), tableName);
            tables.forEach(map -> {
                KeyValueModelVo dataMap = new KeyValueModelVo();
                dataMap.setKey(map.getColumnName());
                dataMap.setDataType(map.getTypeName() + "(" + map.getColumnSize() + ")");
                dataMap.setValue(map.getRemarks());
                resultList.add(dataMap);
            });
        } catch (SQLException sqlException) {
            logger.info("获取数据库信息异常，请检查调用参数。", sqlException);
        }
        return resultList;
    }

    @Override
    public List<KeyValueModelVo> getAllFieldByTableName(CfgDb cfgdb, String tableName) {
        List<KeyValueModelVo> strResult = new ArrayList<KeyValueModelVo>();
        if (DataBaseType.MongoDB.equals(cfgdb.getDbVersion())) {
            MongoDBUtil instance = MongoDBUtil.getInstance();
            QueryInfo queryInfo = new QueryInfo();
            queryInfo.setDb(instance.createConnection(cfgdb.getDbUrl()));
            queryInfo.setCollectionName(tableName);
            //BasicDBObject fields = instance.getFields(queryInfo);
            Map<String, String> orderColumnObject = instance.getFirstElementNameAndTypeInCollection(queryInfo);
            List<KeyValueModelVo> finalStrResult = strResult;
            orderColumnObject.forEach(
                    (columnName, columnType) -> {
                        KeyValueModelVo keyValueModelVo = new KeyValueModelVo();
                        keyValueModelVo.setKey(columnName);
                        keyValueModelVo.setDataType(columnType);
                        finalStrResult.add(keyValueModelVo);
                    });
            instance.close();
        } else if (DataBaseType.KUDU.equals(cfgdb.getDbVersion())) {
            strResult = kuduService.getAllFieldByTableName(cfgdb.getDbName(), tableName);
        } else {
            strResult = getColumnsByDriver(cfgdb, tableName);
        }
        return strResult;
    }

    @Override
    public List<Map<String, Object>> getSqlResult(CfgDb cfgdb, String strsql) {
        List<Map<String, Object>> res = null;
        if (DataBaseType.MongoDB.equals(cfgdb.getDbVersion())) {
            MongoDBUtil instance = MongoDBUtil.getInstance();
            QueryInfo queryInfo = instance.sql2MongoQuery(strsql);
            if (queryInfo == null) {
                return null;
            }
            MongoDatabase db = instance.createConnection(cfgdb.getDbUrl());
            queryInfo.setDb(db);
            res = instance.findAll(queryInfo);
            instance.close();
        } else if (DataBaseType.KUDU.equals(cfgdb.getDbVersion())) {
            res = hiveService.selectTable(strsql);
        } else {
            res = executeSqlMapper.executequerySql(strsql);
        }

        return res;
    }

    /*
     * ModifyBy Cocoliu 2017/10/17
     * 增加返回值内容包括返回状态，以及返回错误信息
     * @see com.elitel.frame.business.service.DataBaseMetadataService#testDataSource(com.elitel.frame.main.entity.CfgDb)
     */
    @Override
    public boolean testDataSource(CfgDb cfgdb) {
        boolean testConnectionFlag = false;
        try {
            if (DataBaseType.MongoDB.equals(cfgdb.getDbVersion())) {
                MongoDBUtil.getInstance().createConnection(cfgdb.getDbUrl());
                testConnectionFlag = true;
            } else {
                testConnectionFlag = getConnection(cfgdb);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return testConnectionFlag;
    }

    @Override
    public Map<String, String> getPkNameMap(CfgDb cfgdb, List<String> tableNameList) {
        switch (cfgdb.getDbVersion()) {
            case DataBaseType.MongoDB:
                return new HashMap<String, String>();
            case DataBaseType.KUDU:
                return kuduService.getPrimaryKeyMap(cfgdb.getDbName());
            default:
                return getPrimaryKeysMapByDriver(cfgdb, tableNameList);
        }
    }

    @Override
    public Map<String, List<KeyValueModelVo>> getAllFieldMap(CfgDb cfgdb, List<String> tableNameList) {
        Map<String, List<KeyValueModelVo>> strResult = new HashMap<>();

        switch (cfgdb.getDbVersion()) {
            case DataBaseType.MongoDB:
                List<Map<String, String>> allTableByDataSource = getAllTableByDataSource(cfgdb);
                Map<String, List<KeyValueModelVo>> mgMap = new HashMap<>();
                for (Map<String, String> map : allTableByDataSource) {
                    String table_name = map.get("table_name");
                    List<KeyValueModelVo> mgList = getAllFieldByTableName(cfgdb, table_name);
                    mgMap.put(table_name, mgList);
                }
                strResult = mgMap;
                break;
            case DataBaseType.KUDU:
                strResult = kuduService.getAllFieldMap(cfgdb.getDbName());
                break;
            default:
                strResult = getColumnsMapByDriver(cfgdb, tableNameList);
        }
        return strResult;
    }

    @Override
    public List<Map<String, Object>> getDBTableRecordCount(CfgDb cfgdb, String tableName) {

        switch (cfgdb.getDbVersion()) {
            case DataBaseType.SQLSERVER:
                return dataBaseMetadataMapper.getSQLServerTableCount(tableName);
            case DataBaseType.ORACLE:
            case DataBaseType.DM:
                return dataBaseMetadataMapper.getOracleTableCount(tableName);
            case DataBaseType.POSTGRESQL:
            case DataBaseType.TBASE:
                if ("true".equals(environment.getProperty("datasource.pg.distribute.statistic"))) {
                    return countDatabaseTableRecord(cfgdb, tableName);
                }
                return dataBaseMetadataMapper.getPostgresTableCount(StringUtils.isNotBlank(cfgdb.getDbSchema()) ? cfgdb.getDbSchema() : "public", tableName);
            case DataBaseType.MYSQL:
                return dataBaseMetadataMapper.getMySQLTableCount(cfgdb.getDbName(), tableName);
            case DataBaseType.KUDU:
                return kuduService.getKuduTableCount(cfgdb.getDbKey(), tableName);
            default:
        }
        return null;
    }

    private List<Map<String, Object>> countDatabaseTableRecord(CfgDb cfgdb, String tableName) {
        String querySQLFormat = "select '%s' as table_name, count(1) as table_count from %s";
        if (StringUtils.isNotBlank(tableName)) {
            return executeSqlMapper.executequerySql(String.format(querySQLFormat, tableName, tableName));
        }

        //获取该数据源配置的所有表
        List<CfgTable> listBydbkey = extCfgTableMapper.getListBydbkey(cfgdb.getDbKey());
        List<Map<String, Object>> resMaps = new ArrayList<>();
        for (CfgTable cfgTable : listBydbkey) {
            String dtName = cfgTable.getDtName();
            List<Map<String, Object>> maps = executeSqlMapper.executequerySql(String.format(querySQLFormat, dtName, dtName));
            if (!maps.isEmpty()) {
                resMaps.add(maps.get(0));
            }
        }
        return resMaps;
    }

    private Map<String, List<KeyValueModelVo>> getColumnsMapByDriver(CfgDb cfgdb, List<String> tableNameList) {
        try {
            //获取所有列名
            Connection connection = dataSource.getConnection();
            List<MetaPrimaryKey> primaryKeys = DatabaseMetaDataHelper.getPrimaryKeysMap(connection, getCurrentSchema(cfgdb), tableNameList);
//            primaryKeys= primaryKeys.stream().filter(o -> o.getTableSchema().equalsIgnoreCase(cfgdb.getDbName())).collect(Collectors.toList());
            List<MetaColumn> tables = DatabaseMetaDataHelper.getColumnsMap(dataSource.getConnection(),
                    getCurrentSchema(cfgdb), primaryKeys, tableNameList);
            Map<String, List<MetaColumn>> tableMap = tables.stream()
//                    .filter(o->o.getTableSchema().equalsIgnoreCase(cfgdb.getDbName()))
                    .collect(Collectors.groupingBy(MetaColumn::getTableName));
            return tableMap.entrySet().stream()
                    .collect(Collectors.toMap(
                            Map.Entry::getKey,
                            entry -> entry.getValue().stream()
                                    .map(o -> {
                                        KeyValueModelVo dataMap = new KeyValueModelVo();
                                        dataMap.setKey(o.getColumnName());
                                        dataMap.setDataType(o.getTypeName());
                                        dataMap.setValue(o.getRemarks());
                                        return dataMap;
                                    }).collect(Collectors.toList())
                    ));
        } catch (Exception sqlException) {
            logger.info("获取数据库信息异常，请检查调用参数。", sqlException);
        }
        return null;
    }

    private Map<String, String> getPrimaryKeysMapByDriver(CfgDb cfgdb, List<String> tableNameList) {
        try {
            //获取主键信息
            List<MetaPrimaryKey> primaryKeys = DatabaseMetaDataHelper.getPrimaryKeysMap(dataSource.getConnection(),
                    getCurrentSchema(cfgdb), tableNameList);
            if (!ObjectUtils.isEmpty(primaryKeys)) {
                Map<String, List<MetaPrimaryKey>> inputMap = primaryKeys.stream()
//                        .filter(o->o.getTableSchema().equalsIgnoreCase(cfgdb.getDbName()))
                        .collect(Collectors.groupingBy(MetaPrimaryKey::getTableName));
                return inputMap.entrySet().stream()
                        .collect(Collectors.toMap(
                                Map.Entry::getKey,
                                entry -> entry.getValue().stream()
                                        .map(MetaPrimaryKey::getColumnName)
                                        .collect(Collectors.joining(","))
                        ));
            }
        } catch (SQLException e) {
            e.printStackTrace();
            logger.info("获取数据库信息异常，请检查调用参数。", e);
        }
        return null;
    }

    public boolean getConnection(CfgDb cfgdb) {
        Connection conn = null;
        boolean bConnect = false;
        String url = "";
        try {
            Class.forName(cfgdb.getDbDrivername());
            url = cfgdb.getDbUrl();
            String user = cfgdb.getDbUser();
            String password = cfgdb.getDbPwd();
            logger.info("jdbc url: " + url + " user:" + user + " password:" + password);
            // 建立连接
            conn = DriverManager.getConnection(url, user, password);
            bConnect = true;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        return bConnect;
    }

    /**
     * 获取数据库链接中的schema
     *
     * @param cfgDb 数据库连接配置信息
     * @return schema名称
     */
    private String getCurrentSchema(CfgDb cfgDb) {
        //MySQL需要特殊处理
        if (MYSQL_DIALECT.equals(cfgDb.getDbVersion())) {
            return cfgDb.getDbName();
        }
        return cfgDb.getDbSchema();
    }

    /**
     * 获取数据库链接中的schema
     *
     * @param url 数据库链接
     * @return schema名称
     */
    private String getSchemaByUrl(String url) {
        if (url.contains("?")) {
            String[] strs = url.substring(url.indexOf("?") + 1).split("&");
            for (String str : strs) {
                String[] pair = str.split("=");
                // TODO 
                if (DatabaseConstant.CURRENT_SCHEMA_DIALECT.equals(pair[0])
                        || DatabaseConstant.SCHEMA_DIALECT.equals(pair[0])
                ) {
                    return pair.length > 1 ? pair[1] : null;
                }
            }
        }
        return null;
    }
}
